package com.my.juc.pool;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author zhupanlin
 * @version 1.0
 * @description: TODO
 * @date 2024/4/27 9:16
 */

/**
 * public ThreadPoolExecutor(int corePoolSize,
 *                               int maximumPoolSize,
 *                               long keepAliveTime,
 *                               TimeUnit unit,
 *                               BlockingQueue<Runnable> workQueue,
 *                               ThreadFactory threadFactory,
 *                               RejectedExecutionHandler handler);
 *     corePoolSize ：核心线程数
 *     maximumPoolSize：最大线程数，救急线程，当核心线程都在执行任务，同时阻塞队列已满，就会创建救急线程执行任务，救急线程数=最大线程数-核心线程数
 *     keepAliveTime：没有任务时，救急线程存活时间，只有救急线程有生存时间，核心线程没有生存时间
 *     unit ：时间单位
 *     BlockingQueue<Runnable> workQueue：多余的任务存放在该阻塞队列中
 *     threadFactory：线程工厂，对线程做定制化操作
 *     RejectedExecutionHandler handler：阻塞队列中存放不了任务，拒绝策略，如果所有的线程都在执行任务，而且阻塞队列已满，就会执行拒绝策略
 */
public class TestPool {
    

    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(1, 1000, TimeUnit.MILLISECONDS, 1,
                (queue, task) -> {
                    // 1.死等
                    //queue.put(task);
                    // 2.带超时等待
                    //queue.offer(task, 500, TimeUnit.MILLISECONDS);
                    // 3. 让调用者放弃任务执行
                    //System.out.println("放弃" + task);
                    // 4.让调用者抛出异常
                    //throw new RuntimeException("任务执行失败" + task);
                    // 5.让调用者自己执行任务
                    task.run();
                });
        for (int i = 0; i < 3; i++) {
            int j = i;
            threadPool.execute(() -> {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println( Thread.currentThread().getName() + "j : " + j);
            });
        }
    }
    
}
@FunctionalInterface
interface RejectPolicy<T> {
    void reject(BlockingQueue<T> queue, T task);
}

class ThreadPool {
    
    private BlockingQueue<Runnable> taskQueue;
    
    // 线程集合
    private HashSet<Worker> workers = new HashSet();
    
    // 核心线程数
    private int coreSize;
    
    // 获取任务的超时时间
    private long timeout;
    
    private TimeUnit timeUnit;
    
    private RejectPolicy<Runnable> rejectPolicy;

    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int capacity, RejectPolicy<Runnable> rejectPolicy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(capacity);
        this.rejectPolicy = rejectPolicy;
    }
    
    public void execute(Runnable task) {
        // 当任务数没有超过 coreSize 时，直接交给worker
        // 如果任务数超过 coreSize，加入任务队列暂存
        synchronized (workers) {
            if (workers.size() < coreSize) {
                Worker worker = new Worker(task);
                System.out.println("新增 worker：" + worker);
                workers.add(worker);
                worker.start();
            } else {
                System.out.println("加入任务队列:" + task );
                // 1.死等
                //taskQueue.put(task);
                // 2.带超时等待
                // 3.放弃任务执行
                // 4.抛出异常
                // 5.让调用者自己执行任务
                taskQueue.tryPut(rejectPolicy, task);
            }
        }
    }

    class Worker extends Thread{
        
        private Runnable task;
        public Worker(Runnable task) {
            this.task = task;
        }
        
        @Override
        public void run() {
            // 执行任务
            // 1.当 task 不为空，执行任务
            // 2.当 task 执行完毕，再接着从任务队列获取任务并执行
            //while (task != null || (task = taskQueue.take()) != null) {
            while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
                try {
                    System.out.println(Thread.currentThread().getName() + "正在执行..." + task);
                    task.run();
                }catch (Exception e) {
                    e.printStackTrace();
                }finally {
                    task = null;
                }
            }
            synchronized (workers) {
                System.out.println(Thread.currentThread().getName() + "执行完毕..." + task);
                workers.remove(this);
            }
        }
        
    }
}

class BlockingQueue<T> {
    // 1.任务队列
    private Deque<T> queue;

    public BlockingQueue(int capacity) {
        this.capacity = capacity;
        this.queue = new ArrayDeque<>(capacity);
    }

    // 2.锁
    private ReentrantLock lock = new ReentrantLock();
    
    // 3.生产者条件变量
    private Condition fullWaitSet = lock.newCondition();
    
    // 4.消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();
    
    // 5.容量
    private int capacity;
    
    // 带超时的阻塞获取
    public T poll(long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            try {
                // 将 timeout 统一转换为纳秒
                long nanos = timeUnit.toNanos(timeout);
                while (queue.isEmpty()) {
                    // 返回的是剩余时间
                    if (nanos <= 0) {
                        return null;
                    }
                    nanos = emptyWaitSet.awaitNanos(nanos);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }
        
    }
    
    // 阻塞获取
    public T take() {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }
    }
    
    // 阻塞添加
    public void put(T element) {
        lock.lock();
        try {
            while (queue.size() == capacity) {
                try {
                    System.out.println("等待加入任务队列" + element);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            queue.addLast(element);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }
    
    // 带超时时间的阻塞添加
    public boolean offer(T task, long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            long nanos = timeUnit.toNanos(timeout);
            while (queue.size() == capacity) {
                try {
                    System.out.println("等待加入任务队列" + task);
                    if (nanos <= 0) {
                        return false;
                    }
                    fullWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            queue.addLast(task);
            emptyWaitSet.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }
    
    // 获取大小
    public int size() {
        lock.lock();
        try {
            return queue.size();
        } finally {
            lock.unlock();
        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
        lock.lock();
        try {
            // 判断队列是否已满
            if (queue.size() == capacity) {
                rejectPolicy.reject(this, task);
            } else {
                System.out.println("加入任务队列" + task);
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        } finally {
            lock.unlock();
        }
    }
}
